package disruptor;

import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.Executors;

/**
 * @author dushougudu
 */
public class DisruptorQueueFactory {
    /**
     * 创建"点对点模式"的操作队列，即同一事件会被一组消费者其中之一消费
     *
     * @param queueSize
     * @param isMoreProducer
     * @param consumers
     * @param <T>
     * @return
     */
    public static <T> DisruptorQueue<T> getWorkPoolQueue(int queueSize, boolean isMoreProducer,
                                                         AbstractConsumer<T>... consumers) {
        Disruptor<ObjectEvent<T>> disruptor = new Disruptor(new ObjectEventFactory(),
                queueSize, Executors.defaultThreadFactory(),
                isMoreProducer ? ProducerType.MULTI : ProducerType.SINGLE,
                new SleepingWaitStrategy());
        disruptor.handleEventsWithWorkerPool(consumers);
        return new DisruptorQueue(disruptor);
    }


    /**
     * 创建"发布订阅模式"的操作队列，即同一事件会被多个消费者并行消费
     *
     * @param queueSize
     * @param isMoreProducer
     * @param consumers
     * @param <T>
     * @return
     */
    public static <T> DisruptorQueue<T> getHandleEventsQueue(int queueSize, boolean isMoreProducer,
                                                             AbstractConsumer<T>... consumers) {
        Disruptor<ObjectEvent<T>> disruptor = new Disruptor(new ObjectEventFactory(),
                queueSize, Executors.defaultThreadFactory(),
                isMoreProducer ? ProducerType.MULTI : ProducerType.SINGLE,
                new SleepingWaitStrategy());
        disruptor.handleEventsWith(consumers);
        return new DisruptorQueue(disruptor);
    }

}
